package com.tt.job.executor;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.tt.job.api.Job;
import com.tt.job.api.JobContext;
import com.tt.job.api.JobResult;
import com.tt.job.api.log.JobLoggerFactory;
import com.tt.job.api.type.dataflow.DataFlowJob;
import com.tt.job.api.type.script.ScriptJob;
import com.tt.job.api.type.simple.SimpleJob;
import com.tt.job.common.util.JsonUtil;
import com.tt.job.container.JobContainer;
import com.tt.job.util.SpringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.List;
import java.util.concurrent.ScheduledThreadPoolExecutor;

/**
 * 所有任务消息处理分发类
 */
public class JobMessageConcurrentlyExecutor implements MessageListenerConcurrently {
    private final static Logger logger = LoggerFactory.getLogger(JobMessageConcurrentlyExecutor.class);
    private final ScheduledThreadPoolExecutor executorService;
    private final Class<? extends Job> jobClass;
    private final String jobClassName;
    private final Integer maxReconsumeTimes;

    public JobMessageConcurrentlyExecutor(ScheduledThreadPoolExecutor executorService, Job job, Integer maxReconsumeTimes) {
        this.executorService = executorService;
        this.jobClass = job.getClass();
        this.jobClassName = job.getClass().getName();
        this.maxReconsumeTimes = maxReconsumeTimes;
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        for (MessageExt msg : msgs) {
            String jobClass = msg.getTags().replace("-", ".");
            if (!jobClass.equals(this.jobClassName)) {
                logger.error("订阅的job:{} 和消息返回的job:{} 信息不一致，跳过！msg:{}", this.jobClassName, jobClass, new String(msg.getBody()));
                //确认这样是否能够解决问题
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            //
            JobContext jobContext = JsonUtil.decode(msg.getBody(), JobContext.class);
            if (jobContext == null) {
                //
                logger.error("消息反序列化失败，跳过消息！msg:{}", new String(msg.getBody()));
                continue;
            }

            JobLoggerFactory.LOGGER_FACTORY.set(JobLoggerFactory.getLogger(jobContext.getSchedule().getJobClass()));
            logger.debug("开始处理任务消息: msgId:{}, queneId:{} job:{}", msg.getMsgId(), msg.getQueueId(), jobClass);
            //重新从spring中获取job实例（针对prototype此次会新建一个job对象）
            Job job;
            JobContainer container = JobContainer.getContainer(this.jobClassName);
            ConsumeConcurrentlyStatus result;
            if (container != null) {
                logger.debug("在容器中运行任务: job:{}, container:{}, Path:{}", jobClassName, container.getName(), container.getPath());
                job = container.getBean(this.jobClass);
                //切换当前线程的classLoader
                Thread.currentThread().setContextClassLoader(container.getClassLoader());
                result = executeJob(msg, jobClass, jobContext, job);
                Thread.currentThread().setContextClassLoader(JobMessageConcurrentlyExecutor.class.getClassLoader());
            } else {
                logger.debug("非容器运行任务：job:{}", jobClassName);
                job = SpringUtils.getBean(this.jobClass);
                result = executeJob(msg, jobClass, jobContext, job);
            }
            if (result == ConsumeConcurrentlyStatus.RECONSUME_LATER) {
                return result;
            }
        }
        logger.debug("执行任务成功 job:{}", jobClassName);
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    private ConsumeConcurrentlyStatus executeJob(MessageExt msg, String jobClass, JobContext jobContext, Job job) {
        if (job == null) {
            //Spring bean 未初始化好，重新处理
            logger.error("获取job失败，等待重试");
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
        //判断如果是定时调度任务，并且此次调度
        Date now = new Date();
        if (jobContext.getNextFireTime() != null && jobContext.getNextFireTime().before(now)) {
            //
            logger.info("TriggerSchedule的后续调度已到达，无需处理当前任务. jobClass:{}, nextFireTime:{}, now:{}", jobClass, jobContext.getNextFireTime(), now);
            return null;
        }
        JobResult response = JobResult.FAILED;
        try {
            if (job instanceof DataFlowJob) {
                //实现dataflow 类型的任务，使用多线程加速任务执行效率
                response = new DataFlowJobMessageExecutor(executorService, (DataFlowJob) job).execute(jobContext);
            } else if (job instanceof ScriptJob) {
                //script job TODO
                logger.error("script job is not supported!");
            } else if (job instanceof SimpleJob) {
                //simple job 简单任务直接处理掉
                //通过Spring获取下job实例
                response = ((SimpleJob) job).execute(jobContext);
            }
            if (msg.getReconsumeTimes() >= maxReconsumeTimes) {
                logger.error("任务执行失败超过次数上限，跳过消息。context:{}", jobContext);
                return null;
            }
            if (response == JobResult.FAILED) {
                logger.error("执行job失败， job:{}, response:{}", jobContext.getJobClass(), response);
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        } catch (Throwable e) {
            logger.error("执行job异常， job:{}", jobContext.getJobClass(), e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
